package org.example.rocketmq.msg_type.filter;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class ConsumerFilterSql {
    public static void main(String[] args) throws Exception {
        consumer();
    }

    /*
    注意：
    1. 只有推模式的消费者可以使用SQL过滤，拉模式是用不了
    2. 另外消息过滤是在 Broker 端进行的，提升网络传输性能，但是 broker 服务会比较繁忙。（consumer 将过滤条件推送给 broker 端）
     */
    /**
     * 顺序消息值生产者局部有序发送到一个 queue, 但多个 queue 之间是全局无序的。
     * 顺序消息消费者样例： 通过 MessageListenerOrderly 消费者每次读取消息都只从一个 queue 中获取（通过加锁的方式实现）
     *
     * 单机情况下在broker.conf中添加
     * #支持sql92
     * enablePropertyFilter=true
     *
     * 集群情况下修改:broker-m.conf和broker-s.conf。然后重启broker即可。
     */
    public static void consumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_order_group_01");
        consumer.setNamesrvAddr("rocketmq.jun.cn:9876");
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe("TopicTest", MessageSelector.bySql("TAGS is not null and TAGS in ('TagA', 'TagC') and (baiLi is not null and baiLi between 0 and 2)"));

        consumer.registerMessageListener(new MessageListenerOrderly() { //  顺序消费
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(i + "_消息消费成功_" + new String(list.get(i).getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });


        /*consumer.registerMessageListener(new MessageListenerConcurrently() { // 并发消费
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(i + "_消息消费成功_" + new String(list.get(i).getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });*/

        consumer.start();
        System.out.printf("Consumer started %n");
    }
}
